/**
   \file  thread_pool.hh
   \brief Defines a generic thread pool.

   This file implements a generic thread pool that can be used by other
   parts of libgist. It is not meant to be a bulletproof, industrial
   strength thread pool; rather, it is a quick (but, hopefully, not too
   dirty) thread pool implementation designed specifically for use by
   libgist (i.e., not a general-purpose class for wide deployment).

   DEVNOTE: Ideally, we would like to use a thread pool implementation
   provided by a readily available library. Unfortunately, at this time
   (circa February 2012), Boost.Thread does not implement a thread pool
   and Boost.ThreadPool is not yet a part of Boost.
*/

/*
   This file is part of libgist.

   libgist is free software; you can redistribute it and/or modify it
   under the terms of the GNU General Public License as published by the
   Free Software Foundation; either version 2 of the License, or (at your
   option) any later version.

   libgist is distributed in the hope that it will be useful, but WITHOUT
   ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
   FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
   for more details.

   You should have received a copy of the GNU General Public License
   along with libgist; if not, write to the Free Software Foundation,
   Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
*/

/*
   REVISION HISTORY

   $HeadURL: http://libgist.googlecode.com/svn/branches/lazebnik_dev/src/thread_pool.hh $
   $Id: thread_pool.hh 160 2012-02-21 22:28:57Z libgist@gmail.com $
*/

#ifndef GIST_THREAD_POOL_DOT_HH
#define GIST_THREAD_POOL_DOT_HH

//------------------------------ HEADERS --------------------------------

// Boost
#include <boost/thread/thread.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/locks.hpp>
#include <boost/thread/shared_mutex.hpp>
#include <boost/thread/mutex.hpp>

#include <boost/signals2/signal.hpp>
#include <boost/bind.hpp>

// Standard C++
#include <sstream>
#include <queue>

//---------------------------- NAMESPACE --------------------------------

namespace gist {

//------------------------- THREAD POOL CLASS ---------------------------

/**
   \brief A quick (but not too dirty) thread pool.

   This class provides a relatively straightforward abstraction for
   queueing multiple tasks to be executed in parallel.

   The task type for the thread_pool class is expected to be a callable
   (i.e., a function or function object). Clients can pass whatever
   arguments they like to these tasks. Furthermore, it is their
   responsibility to ensure that the tasks are thread-safe.

   DEVNOTE: Ideally, we would like to use a thread pool implementation
   provided by a readily available library. Unfortunately, at this time
   (circa February 2012), Boost.Thread does not implement a thread pool
   and Boost.ThreadPool is not yet a part of Boost.
*/
template<typename task>
class thread_pool {
   /// The pool of threads is implemented using a boost::thread_group.
   boost::thread_group m_threads ;

   /// When a client has a large job that can be parallelized, it
   /// should break that job into smaller, independent steps or tasks
   /// and then queue each of those sub-tasks with the thread pool.
   /// Each thread in the pool will retrieve the next task from the
   /// queue and execute it independently of the other threads.
   ///
   /// A task is simply a function or function object whose arguments are
   /// specified by clients. If these tasks need to maintain state, they
   /// should probably be implemented as function objects. If they need
   /// to share variables and need synchronization, clients should take
   /// care of the details of mutual exclusion, etc.
   ///
   /// This data member holds the queue of tasks described above.
   std::queue<task> m_tasks ;

   /// Since all the threads in the pool access the task queue, we
   /// have to synchronize access to it with a mutex. Moreover, to
   /// ensure that the threads don't uselessly spin when the task
   /// queue is empty, we use a condition variable to have each thread
   /// block when there are no tasks to execute.
   //@{
   boost::mutex m_mutex ;
   boost::condition_variable m_cond ;
   //@}

   /// These variables allow the main (or client) thread that launches
   /// multiple parallel tasks to wait for all those tasks to
   /// complete.
   ///
   /// NOTE: The thread pool cannot, in fact, know that all of a
   /// particular client's parallel tasks are done. This condition
   /// variable actually indicates that the pool's task queue is empty
   /// and that it has finished executing the last task it had queued.
   /// However, by first adding all the parallel tasks and then
   /// waiting for the pool's task queue to become empty, client
   /// threads can be reasonably confident that all the tasks are, in
   /// fact, done.
   ///
   /// DEVNOTE: It is not enough to check that the task queue is empty
   /// to be able to notify the main/client thread that all tasks have
   /// been executed. We must also ensure that the last task has
   /// finished executing. Otherwise, the main thread may be notified
   /// prematurely and continue executing, potentially deleting
   /// resources still in use by the last task, which may still be
   /// working. When that happens, hello Segfault City!
   //@{
   int m_num_tasks ;
   boost::mutex m_wait_mutex ;
   boost::condition_variable m_wait_cond ;
   //@}

   /// When a thread pool object goes out of scope, we use this flag to
   /// inform all the threads in the pool to gracefully quit.
   bool m_shutdown ;

   /// The shutdown flag defined above is set in the thread pool's
   /// destructor. The pool's threads themselves only read this flag.
   /// Thus, we use a read-write mutex to synchronize accesses to this
   /// variable.
   boost::shared_mutex m_shutdown_mutex ;

   /// By default, when a Boost.Thread encounters an exception, the
   /// library terminates the application. Instead, we allow clients to
   /// handle the exception in whatever way they see fit using this
   /// signal.
   //@{
   typedef boost::signals2::signal<void (const boost::thread::id&,
                                         const std::string&)>
           thread_exception_handler ;
   thread_exception_handler handle_thread_exception ;
   //@}

public:
   /**
      \brief  Thread pool initialization.
      \param  n Number of threads in the pool.
      \return Pool of n threads waiting to be fed tasks to execute.

      On instantiation, clients should specify how many threads the pool
      should have. The constructor will take care of launching that many
      threads.
   */
   thread_pool(int n) ;

   /**
      \brief  Thread pool clean-up.
      \return Nothing.

      In the thread pool's destructor, we set the shutdown flag and
      then wait for all the threads to exit. This will usually happen
      when the kmeans object goes out of scope, thus, invoking its
      embedded thread pool object's destructor.
   */
   ~thread_pool() ;

   /**
      \brief  Add a task to the thread pool.
      \param  t A function or function object implementing the task.
      \return Nothing.

      This method adds the given function or function object to the
      thread pool's queue of tasks. It will be executed when it
      reaches the front of the queue and when one of the threads in
      the pool becomes available.

      The task function is expected to take no parameters and return
      nothing. If some tasks need to share resources, they should be
      implemented in a way that ensures proper synchronization. That is,
      clients are responsible for implementing tasks so that they are
      thread-safe.
   */
   void add_task(task t) ;

   /**
      \brief  Wait for thread pool's task queue to become empty.
      \return Nothing.

      In general, after a thread has added several parallel tasks to the
      thread pool's task queue, it should wait for all the tasks to
      complete, which it can do with this API.

      The usual pattern for using this thread pool is as follows:

      \code
          for (i = 0; i < n; ++i)
              thread_pool_instance.add_task(whatever) ;
          thread_pool_instance.wait() ;
      \endcode
   */
   void wait() ;

   /**
      \brief  Add a callback for handling thread errors.
      \param  f The callback function for handling thread errors.
      \return Nothing.

      libgist's thread_pool class relies on the Boost.Thread library,
      whose default behaviour on encountering a thread error is to
      terminate the application. To give clients more flexibility in this
      matter, we instead trigger a client-specified callback every time
      we encounter a thread related error.

      The function signature of this callback should be as follows:

      \code
          void f(const boost::thread::id&, const std::string& msg)
      \endcode

      In this callback, clients may handle the error by printing it out
      or taking some other useful and appropriate action.

      Note that some exceptions are fatal and will result in the thread
      that receives them terminating itself. Other threads in the pool,
      however, will continue unimpeded. Non-fatal exceptions will simply
      be reported via the callback and then continue the thread.

      This method adds a thread error handling callback to the thread
      pool's internal list of callbacks.
   */
   void
   add_thread_exception_handler(const thread_exception_handler::slot_type& f) {
      handle_thread_exception.connect(f) ;
   }

   /// \return The number of threads in the pool.
   int size() const {return m_threads.size() ;}

private:
   /**
      \brief  Thread pool's thread function.
      \return Nothing.

      Each thread in the pool does the same thing: wait for a task and
      then execute it. In each iteration of this loop, we also check
      to see if the shutdown flag has been set and, if so, quit the
      loop, thereby, terminating each thread.
   */
   void thread_func() ;
} ;

//-------------------- INITIALIZATION AND CLEAN-UP ----------------------

// Initialization: create all the threads in the pool
template<typename task>
thread_pool<task>::thread_pool(int n)
   : m_num_tasks(0), m_shutdown(false)
{
   using boost::bind ;
   using boost::thread ;

   if (n < 1)
       n = 1 ;
   for (int i = 0; i < n; ++i)
      m_threads.add_thread(new thread(bind(&thread_pool::thread_func, this))) ;
}

// Clean-up: notify all threads in the pool it's time to quit
template<typename task>
thread_pool<task>::~thread_pool()
{
   // Set shutdown flag to indicate to all threads it's time to quit
   // DEVNOTE: Block to automatically acquire and release mutex.
   {
      boost::lock_guard<boost::shared_mutex> write_lock(m_shutdown_mutex) ;
      m_shutdown = true ;
   }

   // Wake up all threads in the pool and wait for them to quit in
   // response to setting above flag.
   m_cond. notify_all() ;
   m_threads.join_all() ;
}

//-------------------------- THREAD POOL API ----------------------------

// Add a new task to pool's queue and notify one of the threads that
// there's a task waiting to be executed.
template<typename task>
void thread_pool<task>::add_task(task t)
{
   // DEVNOTE: Local block to automatically acquire and release mutexes.
   {
      boost::lock_guard<boost::mutex> task_queue_lock(m_mutex) ;
      boost::lock_guard<boost::mutex> task_count_lock(m_wait_mutex) ;
      m_tasks.push(t) ;
      ++m_num_tasks ;
   }
   m_cond.notify_one() ;
}

// API to allow main thread to wait for all tasks currently in the pool
// to complete.
template<typename task>
void thread_pool<task>::wait()
{
   try
   {
      boost::unique_lock<boost::mutex> lock(m_wait_mutex) ;
      while (m_num_tasks > 0)
         m_wait_cond.wait(lock) ;
   }
   catch (boost::thread_resource_error& e)
   {
      std::ostringstream msg ;
      msg << "thread_resource_error while waiting for thread pool to empty: "
          << e.what() ;
      handle_thread_exception(boost::this_thread::get_id(), msg.str()) ;
   }
   catch (boost::thread_interrupted&)
   {
      handle_thread_exception(boost::this_thread::get_id(),
         "interrupted while waiting for thread pool to empty") ;
   }
}

//---------------------- THREAD POOL THREAD FUNC ------------------------

// Each thread in the pool basically waits for tasks to arrive and then
// executes them.
template<typename task>
void thread_pool<task>::thread_func()
{
   boost::thread::id id = boost::this_thread::get_id() ;
   task t ; // task to execute
   for(;;)
   {
      try
      {
         // First check if the thread pool has been shutdown
         // DEVNOTE: Block to automatically acquire and release mutex
         {
            typedef boost::shared_lock<boost::shared_mutex> read_lock ;
            read_lock r(m_shutdown_mutex) ;
            if (m_shutdown)
               break ;
         }

         boost::unique_lock<boost::mutex> lock(m_mutex) ;
         if (m_tasks.empty()) {   // wait for a task to arrive
             m_cond.wait(lock) ;
             if (m_tasks.empty()) // task consumed by another thread
               continue ; // check thread pool shutdown signal
         }

         // Got a task to work on: take it off the queue
         t = m_tasks.front() ;
         m_tasks.pop() ;
      }
      catch (boost::thread_resource_error& e)
      {
         std::ostringstream msg ;
         msg << "fatal thread_resource_error: " << e.what() ;
         handle_thread_exception(boost::this_thread::get_id(), msg.str()) ;
         break ; // fatal error; thread cannot continue
      }
      catch (boost::thread_interrupted&)
      {
         handle_thread_exception(boost::this_thread::get_id(),
                                 "non-fatal thread interruption") ;
         continue ; // not fatal
      }

      // Execute the latest task retrieved from the queue
      t() ;

      // If all tasks have been executed, notify main/client thread that
      // is probably (or at least should be) waiting for this event...
      //
      // DEVNOTE: Not enough to check that task queue is empty. We must
      // confirm that the last task has completed its work. Otherwise,
      // the main/client thread will be notified prematurely and may
      // delete resources still in use by the still extant last task,
      // which, apart from being an incorrect thread synchronization,
      // will also cause a segfault. That's why we need the num tasks
      // counter.
      boost::unique_lock<boost::mutex> lock(m_wait_mutex) ;
      if (--m_num_tasks <= 0) {
         m_num_tasks = 0 ; // don't want this becoming negative
         m_wait_cond.notify_all() ;
      }
   }
}

//-----------------------------------------------------------------------

} // namespace gist

#endif

/* So things look consistent in everyone's emacs... */
/* Local Variables: */
/* indent-tabs-mode: nil */
/* End: */
